package com.we.risk.registerchanneldeprecated.browseinfo.dim;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;

public class TableBrowseOriginDataHive {
    public static final String KFKTABLE = "adm.rp_browse_origin_sort_by_idfa_channel_da";
    // 近30天数据
    public static final String QUERYSQL =
            "select id_combine, "
                    + "unique_channel, "
                    + "earliest_receive_time, "
                    + "latest_receive_time, "
                    + "UNIX_TIMESTAMP(dt, 'yyyy-MM-dd') as dt_time "
                    + "from "
                    + KFKTABLE
                    + " where dt >= '2022-12-14' "
                    + "and dt <= '2023-01-14'";
    public String unique_channel;
    public String id_combine;
    public long dt_time;
    public long earliest_receive_time;
    public long latest_receive_time;

    public TableBrowseOriginDataHive() {}

    public TableBrowseOriginDataHive(
            String unique_channel,
            String id_combine,
            long dt_time,
            long earliest_receive_time,
            long latest_receive_time) {
        this.unique_channel = unique_channel;
        this.id_combine = id_combine;
        this.dt_time = dt_time;
        this.earliest_receive_time = earliest_receive_time;
        this.latest_receive_time = latest_receive_time;
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        tableEnv.useCatalog("cdh2_hive");
        TableConfig tableConf = tableEnv.getConfig();
        Configuration selfConf = new Configuration();
        selfConf.setBoolean("table.exec.hive.infer-source-parallelism", false);
        tableConf.addConfiguration(selfConf);
        Table tableRes = tableEnv.sqlQuery(TableBrowseOriginDataHive.QUERYSQL);
        DataStream<TableBrowseOriginDataHive> inputStream =
                tableEnv.toDataStream(tableRes, TableBrowseOriginDataHive.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .keyBy(
                                new KeySelector<TableBrowseOriginDataHive, String>() {
                                    @Override
                                    public String getKey(TableBrowseOriginDataHive value)
                                            throws Exception {
                                        return value.getId_combine();
                                    }
                                })
                        .flatMap(
                                new RichFlatMapFunction<TableBrowseOriginDataHive, String>() {

                                    @Override
                                    public void flatMap(
                                            TableBrowseOriginDataHive value, Collector<String> out)
                                            throws Exception {
                                        String id_combine = value.getId_combine();
                                        String unique_channel = value.getUnique_channel();
                                        long dt_time = value.getDt_time();
                                        long earliest_recv_time =
                                                value.getEarliest_receive_time() / 1000;
                                        long latest_recv_time =
                                                value.getLatest_receive_time() / 1000;

                                        JSONObject outJson = new JSONObject();
                                        outJson.put("id_combine", id_combine);
                                        outJson.put("unique_channel", unique_channel);
                                        outJson.put("dt_time", dt_time);
                                        outJson.put("label_name", "earliest_recv_time");
                                        outJson.put("label_value", earliest_recv_time);
                                        out.collect(outJson.toString());

                                        outJson.put("label_name", "latest_recv_time");
                                        outJson.put("label_value", latest_recv_time);
                                        out.collect(outJson.toString());
                                    }
                                });
        //                .flatMap(new browseOriginRichFlatMapFunc());

        return out;
    }

    public String getUnique_channel() {
        return unique_channel;
    }

    public String getId_combine() {
        return id_combine;
    }

    public long getDt_time() {
        return dt_time;
    }

    public long getEarliest_receive_time() {
        return earliest_receive_time;
    }

    public long getLatest_receive_time() {
        return latest_receive_time;
    }
}
